package com.mall.promo.mq;

import com.alibaba.fastjson.JSON;
import com.mall.order.dto.CreateSeckillOrderRequest;
import com.mall.promo.cache.CacheManager;
import com.mall.promo.dal.persistence.PromoItemMapper;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import java.nio.charset.Charset;
import java.util.HashMap;
import java.util.Map;

/**
 * @author: lds
 * @clsss: PromoOrderProducer
 * @description:
 */
@Component
public class PromoOrderProducer {
    private TransactionMQProducer producer;

    @Autowired
    PromoItemMapper promoItemMapper;

    @Autowired
    CacheManager cacheManager;
    @Autowired
    RedissonClient redissonClient;

    @PostConstruct
    public void init() {
        producer = new TransactionMQProducer("promo_order_producer");
        producer.setNamesrvAddr("127.0.0.1:9876");
        producer.setTransactionListener(new TransactionListener() {

            @Override
            public LocalTransactionState executeLocalTransaction(Message message, Object arg) {
                Map<String, Long> paramMap = (Map<String, Long>) arg;
                Long psId = paramMap.get("psId");
                Long productId = paramMap.get("productId");
                //扣减库存(当事务消息发送成功，rocket给了反馈)加分布式锁
                //1.先获得一把锁（针对的是某一个秒杀场次中的某个商品）
                String lockKey = "promo_item_stock" + psId + "_" + productId;
                String transactionId = message.getTransactionId();
                RLock lock = redissonClient.getLock(lockKey);
                String localTransactionKey = "promo_order_local_transaction" + "_" + transactionId /*+ "_" + psId + "_" + productId*/;
                lock.lock();//ip+线程id
                try {
                    Integer effectiveRow = promoItemMapper.decreaseStock(productId, psId);
                    //扣减库存失败
                    //每个事务消息transactionId都是唯一的rocketMQ保证,所以可以用它来唯一标识本地的执行结果
                    if (effectiveRow < 1) {

                        //一天过期
                        cacheManager.setCache(localTransactionKey, "fail", 1);
                        return LocalTransactionState.ROLLBACK_MESSAGE;
                    }
                } finally {
                    lock.unlock();
                }


                cacheManager.setCache(localTransactionKey, "success", 1);

                return LocalTransactionState.COMMIT_MESSAGE;
            }

            @Override      //查询本地状态结果
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                String transactionId = messageExt.getTransactionId();
                String localTransactionKey = "promo_order_local_transaction" + "_" + transactionId;
                String s = cacheManager.checkCache(localTransactionKey);
                //字符串为空，事务没有写入redis
                if (s == null || s.trim().isEmpty()) {
                    return LocalTransactionState.UNKNOW;
                }
                if ("success".equals(s)) {
                    return LocalTransactionState.COMMIT_MESSAGE;

                }
                //失败
                return LocalTransactionState.ROLLBACK_MESSAGE;
            }
        });
        try {
            producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }

    /*
    用来发送，秒杀下单的事务消息的方法
     */
    public boolean sendPromoOrderMessage(CreateSeckillOrderRequest request, Long psId, Long productId) {
        String s = JSON.toJSONString(request);
        byte[] bytes = s.getBytes(Charset.forName("utf-8"));
        Message message = new Message();
        message.setTopic("create_promo_order");
        message.setBody(bytes);

        Map<String, Long> paramMap = new HashMap<>();
        paramMap.put("psId", psId);
        paramMap.put("productId", productId);

        TransactionSendResult sendResult = null;

        try {
            sendResult = producer.sendMessageInTransaction(message, paramMap);
        } catch (MQClientException e) {
            e.printStackTrace();
        }
        //获取事务消息对应的本地事务的执行结果

        if (sendResult != null && LocalTransactionState.COMMIT_MESSAGE.equals(sendResult.getLocalTransactionState())) {
            System.out.println("发送成功");
            return true;

        }
        System.out.println("发送失败");
        return false;

    }

}
